【MCP】AIエージェントをSDKのみで実装　「素」のMCP実装を徹底解説 #生成AI – Qiita

import asyncio
import json
import os
import logging
from mcp.client.session import ClientSession
from mcp.client.stdio import StdioServerParameters, stdio_client
from openai import AsyncOpenAI
from dotenv import load_dotenv
from pydantic import AnyUrl, parse_obj_as

# 環境変数の読み込み
load_dotenv()
env_vars = os.environ.copy()

# ロギング設定
logging.basicConfig(
    level=logging.DEBUG,
    format="%(asctime)s [MCP-CLIENT-BALL] %(message)s"
)
logger = logging.getLogger("mcp-client-ball")

# MCPサーバーの設定（グローバル変数として定義）
SERVER_COMMAND = os.getenv("PYTHON_PATH", "python")  # デフォルトは "python"
SERVER_SCRIPT = "./server_ball/mcp_server_ball.py"
SERVER_PARAMS = StdioServerParameters(
    command=SERVER_COMMAND,
    args=[SERVER_SCRIPT],
    env=env_vars
)

LLM_MODEL_NAME ="gpt-4o"

# LLMに投入するシステムメッセージ
LLM_SYSTEM_MESSAGE = "あなたは、ユーザーの依頼を受けて、玉が入った箱の管理をするAIです。ユーザーの依頼に応じて、玉に文字を書いて箱に入れたり、その箱に特定の文字が書かれた玉が入っているか確認したり、玉の数を確認したりします。玉を箱に入れたら、何をしたのか報告してください。玉が入った箱の管理とは関係ないプロンプトにはツールを適用しないで、無視してください。"

# グローバル変数で状態を保持（各関数は自己完結するが、状態は共有）
available_tools = [] # MCP形式でのツール一覧
available_tools_gpt = [] # OpenAI形式でのツール一覧
available_resources = [] # リソース付属情報一覧
conversation_history = [] # LLMとMCPクライアントとのやりとりの履歴
is_initialized = False

def initialize_client():
    """
    MCPクライアントの初期化、MCPサーバーからツール一覧と各リソースの付属情報一覧を取得する
    """
    global available_tools, available_tools_gpt, available_resources, is_initialized

    if is_initialized:
        logger.info("既に初期化済みのため、再初期化をスキップします")
        return True

    # MCPクライアント初期化処理
    async def async_init():
        global available_tools, available_tools_gpt, available_resources

        # MCPサーバーとMCPで通信
        # MCPサーバーのためのプロセスを開始、MCPサーバーをスクリプト実行、MCPサーバーとの標準入出力通信ルートを構築
        async with stdio_client(SERVER_PARAMS) as (read, write): 
            # MCPサーバーとのMCP通信セッションを構築
            async with ClientSession(read, write) as session:
                # MCPサーバーとの初期化ハンドシェイク
                await session.initialize()
                
                # ツール一覧取得                
                list_tool_result = await session.list_tools()
                # 戻り値の型は、mcp.types.ListToolsResult
                available_tools = list_tool_result.tools
                # MCPサーバーから取得したMCP形式のツール一覧をOpenAI形式に変換
                available_tools_gpt = [
                    {
                        "type": "function",
                        "name": tool.name,
                        "description": tool.description,
                        "parameters": tool.inputSchema
                    }
                    for tool in available_tools
                ]

                # 各リソースの付属情報の一覧取得
                # 本機能では、MCPサーバーのリソースの増減が無く常に一定なので、ここで取得
                # MCPサーバーのリソースの増減がある場合は、ここではなく、必要に応じて取得
                list_resource_result = await session.list_resources()
                # 戻り値の型は、mcp.types.ListResourcesResult
                available_resources = list_resource_result.resources                          

            # async with ClientSessionを抜け、MCPサーバーとのMCP通信セッションが自動的に遮断される
        # async with stdio_clientを抜け、MCPサーバーとの標準入出力通信ルートの遮断とサーバープロセス終了が自動で行われる
    
        return True

    try:
        result = False
        result = asyncio.run(async_init())
        is_initialized = result
        return result
    except Exception as e:
        logger.error(f"MCPクライアント（玉の箱の管理）初期化処理中にエラー: {type(e).__name__}: {e}")
        return False

def process_user_prompt(user_input: str) -> str:
    """
    ユーザー依頼を処理し、GPTにツールを選択させ、
    必要なツール呼び出し・最終応答までを行う。
    """
    
    if not is_initialized:
        logger.info("MCPクライアント（玉の箱の管理）が未初期化のため、初期化を実行")
        initialize_client()
        
    # ユーザーの依頼を会話履歴に追加
    conversation_history.append({"role": "user", "content": user_input}) 
    
    async def async_process():
        try:
            # clientインスタンスを生成
            client = AsyncOpenAI(api_key=os.environ.get("OPENAI_API_KEY"))
            
            # GPTが、ユーザー依頼に対応するのに適切なツールをツール一覧から選択、引数も生成
            response = await client.responses.create(
                model=LLM_MODEL_NAME,
                tools=available_tools_gpt,
                input=[{"role": "system", "content": LLM_SYSTEM_MESSAGE}] + conversation_history
            )
            # response.outputはlist
            
            # GPTからの返答responseは、「自然言語でのメッセージ」or「選択したツール」のどちらか
            response_out_type = response.output[0].type
            # response_out_typeが"message"なら「自然言語でのメッセージ」　→　ツールは選択せず
            # response_out_typeが"function_call"なら「選択したツール」　→　以下の、MCPサーバーにツールを実行させる処理に入る

            # MCPサーバーとMCPで通信
            # MCPサーバーのためのプロセスを開始、MCPサーバーをスクリプト実行、MCPサーバーとの標準入出力通信ルートを構築
            async with stdio_client(SERVER_PARAMS) as (read, write):

                # MCPサーバーとのMCP通信セッションを構築
                async with ClientSession(read, write) as session:
                    # MCPサーバーとの初期化ハンドシェイク
                    await session.initialize()

                    while response_out_type == "function_call":
                        # GPTが何らかのツールを選択
                        tool_calls = response.output                

                        # GPTが選択したツールを順次実行
                        for call in tool_calls:
                            
                            if call.type != "function_call":                                
                                continue
                                
                            # 会話履歴にツール選択結果を追加
                            # callの型は、openai.types.responses.response_function_tool_call.ResponseFunctionToolCall
                            conversation_history.append(call)
                            # GPTが選択したツールの名前を取得
                            tool_name = call.name
                            # GPTが生成したツールの引数を取得
                            arguments = json.loads(call.arguments)
                            # GPTより採番されたcall_idを取得
                            call_id = call.call_id
                            # MCPサーバーにツールを実行させる
                            result_content = await session.call_tool(tool_name, arguments)

                            # MCPサーバーでのツール実行結果のjson文字列（欲しい答えはこれ）を戻り値result_contentから抽出
                            # 戻り値result_contentはmcp.types.CallToolResult型
                            # result_content.contentはlistで、MCPサーバーのcall_toolのシグネチャの戻り値の型list[Union[mcp.types.TextContent, mcp.types.ImageContent, mcp.types.EmbeddedResource]]そのもの
                            # MCPサーバー「mcp_server_ball」のcall_toolの実行結果はmcp.types.TextContentオブジェクトが1個のみのlistであることがわかっているので、以下のような簡素な抽出法でよい。
                            result_text = result_content.content[0].text

                            # JSONパースできる場合、status確認（ツール実行時にエラーが起こったかどうか）
                            # しかし、is_errorフラグそのものには使いみちは無いので、やらなくてもいいかも。
                            # ツール実行時のエラーには、玉に書く文字列（玉のキー文字列）の重複、というのがあり、必ずしもこのユーザー依頼対応の順次処理を止めるべきものとは限らない。
                            try:
                                result_json = json.loads(result_text)
                                is_error = (result_json.get("status") == "error")
                                e_msg_tool = result_json.get("data").get("error") if is_error else ""
                            except Exception:
                                logger.warning("結果文字列をJSONに変換できず")
                                # ツール実行結果文字列のJSON変換に失敗しているので、ツール実行時にエラー発生とみなす
                                is_error = True
                                e_msg_tool = "ツール実行時に不明なエラー"
                                
                            if is_error:
                                # ツール実行時にエラーが起こっても処理は止めない。
                                # ツール実行のこのターン（forループの）でエラーが起こっても、次のターンに影響を及ぼさない。
                                logger.warning(f"ツール実行時のエラー:{e_msg_tool}")
                                
                            # 会話履歴にツール実行結果を追加
                            conversation_history.append({
                                "type": "function_call_output",
                                "call_id": call_id,
                                "output": result_text
                            })
                        # for　ループを抜けた。GPTにより選択されていた全ツールの実行を完了した。

                        # 最終応答の生成 or 次のFunction callのループ
                        # ＜最終応答の生成＞
                        # ユーザー依頼が、実行すべき全ツールを確定させるものの場合
                        # （「玉に鈴木と書いて箱に入れて。箱に今玉は何個ある？」）
                        # このresponseのcreateでは、全ツールの実行が完了したので、最終応答の生成となる。
                        # response.output[0].typeは"message"になる。
                        # ＜次のFunction callのループ＞
                        # ユーザー依頼が、実行すべきツールを全確定させない（あるツールの実行で、次のツールが決まる）ものの場合
                        # （「箱に鈴木と書いてある玉が無ければ追加して」）
                        # もし、そのツールの実行結果が、さらなるツールの実行を要求する時（箱に鈴木と書いてある玉が無かった）、
                        # このresponseのcreateでは、最終応答の生成ではなく、次のFunction callのループとなる。
                        # response.output[0].typeは"function_call"になる。
                        response = await client.responses.create(
                            model=LLM_MODEL_NAME,
                            tools=available_tools_gpt,
                            input=[{"role": "system", "content": LLM_SYSTEM_MESSAGE}] + conversation_history
                            )
                            
                        response_out_type = response.output[0].type
                        # response_out_typeが"message"なら、responseには最終応答が入っている。whileを抜ける。
                        # response_out_typeが"function_call"なら、responseには次に実行すべきツールが入っている。whileループをもう一回行う。

                    # while response_out_type == "function_call"　ループを抜けた。
            
                # async with ClientSessionを抜け、MCPサーバーとのMCP通信セッションが自動的に遮断される
            # async with stdio_clientを抜け、MCPサーバーとの標準入出力通信ルートの遮断とサーバープロセス終了が自動で行われる
            
            #最終応答の抽出
            final_text = response.output_text
            
            # 会話履歴に最終応答を追加
            conversation_history.append({"role": "assistant", "content": final_text})
            
            return final_text
            
        except Exception as e:
            logger.error(f"ユーザー依頼の処理中にエラー発生: {type(e).__name__}: {e}")
            return f"[MCP] エラーが発生しました: {str(e)}"
            
    try:
        return asyncio.run(async_process())
    except Exception as e:
        logger.error(f"asyncio実行エラー: {type(e).__name__}: {e}")
        return f"[MCP] 実行エラー: {str(e)}"


def read_resource_by_uri(uri: str) -> str:
    """
    MCPサーバーから指定URIのリソース1個を取得する
    """
    
    if not is_initialized:
        logger.info("MCPクライアント（玉の箱の管理）が未初期化のため、初期化を実行")
        initialize_client()

    async def async_read():
        try:
            # MCPサーバーとMCPで通信
            async with stdio_client(SERVER_PARAMS) as (read, write):
                async with ClientSession(read, write) as session:
                    await session.initialize()
                    
                    # MCPサーバーからリソースを取得
                    uri_ob = parse_obj_as(AnyUrl, uri)
                    resource_data = await session.read_resource(uri_ob)
                    # resource_dataはmcp.types.ReadResourceResult型
                    # 欲しいリソースは文字列、resource_dataから抽出
                    return resource_data.contents[0].text
                    
        except Exception as e:
            logger.error(f"リソース取得エラー: {type(e).__name__}: {e}")
            return f"[MCP] リソース取得エラー: {str(e)}"

    try:
        return asyncio.run(async_read())
    except Exception as e:
        logger.error(f"asyncio実行エラー: {type(e).__name__}: {e}")
        return f"[MCP] 実行エラー: {str(e)}"